package com.base.component.aws.sqs;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Service;

import com.amazonaws.services.sqs.AmazonSQSAsync;
import com.amazonaws.services.sqs.model.Message;
import com.amazonaws.services.sqs.model.ReceiveMessageRequest;
import com.amazonaws.services.sqs.model.ReceiveMessageResult;
import com.amazonaws.services.sqs.model.SendMessageRequest;
import com.amazonaws.services.sqs.model.SendMessageResult;
import com.base.core.context.utils.LoggerWrapper;
import com.base.core.context.utils.SpringUtils;
import com.gitee.magic.core.exception.ApplicationException;
import com.gitee.magic.core.utils.StringUtils;
import com.gitee.magic.core.utils.codec.Md5;
import com.gitee.magic.framework.head.converter.SerializeConverterEditor;

/**
 * @author StartWei
 */
@Service
public class MqMessageTemplate implements ApplicationRunner {
	
	protected final Logger LOGGER = LoggerFactory.getLogger(getClass());

	@Autowired
	private AmazonSQSAsync amazonSqsAsync;
	
	@SuppressWarnings("rawtypes")
	@Autowired
	private List<MqMessageListener> messageListeners;
	
	private ExecutorService threadPool = Executors.newCachedThreadPool();
	
	
	public SendMessageResult sendMessage(String topic, String tag, String groupId, Object messageBody) {
		return sendMessage(topic, tag, groupId, messageBody,false);
	}
	
	/**
	 * 发送消息
	 * @param topic	主题
	 * @param tag	标签
	 * @param groupId 消息组按先进先出原则同组前面消息未处理会导致后续消息排队,不同消息建议不要在同一组
	 * @param messageBody 消息对象
	 * @param duplicates 丢弃重复消息
	 * @return
	 */
	public SendMessageResult sendMessage(String topic, String tag, String groupId, Object messageBody,Boolean duplicates) {
		String queueUrl = amazonSqsAsync.getQueueUrl(topic).getQueueUrl();
		MqMessage message=new MqMessage();
		message.setTag(tag);
		message.setMessage(messageBody);
		
		SerializeConverterEditor editor = new SerializeConverterEditor(Object.class);
		editor.restore(message);
		String body = String.valueOf(editor.getValue());
		
		SendMessageRequest sendMessageRequest = new SendMessageRequest(queueUrl, body)
				.withMessageGroupId(groupId)
				.withMessageDeduplicationId(duplicates?Md5.md5(body):StringUtils.random());
		return amazonSqsAsync.sendMessage(sendMessageRequest);
	}

	@SuppressWarnings("rawtypes")
	@Override
	public void run(ApplicationArguments args) throws Exception {
		Map<String,MqMessageListener> messageListenerMap=new HashMap<>();
		for (MqMessageListener listener : messageListeners) {
			if (!listener.getClass().isAnnotationPresent(MqListener.class)) {
				continue;
			}
			MqListener mqListener = listener.getClass().getAnnotation(MqListener.class);
			String topicName=SpringUtils.getConfigValue(mqListener.topic());
			String tagName=SpringUtils.getConfigValue(mqListener.tag());
			String key=topicName+"_"+tagName;
			if(messageListenerMap.containsKey(key)) {
				throw new ApplicationException("mq message queue key:"+key+" exist");
			}
			messageListenerMap.put(key, listener);
		}
		for(String topic:messageListenerMap.keySet()) {
			threadPool.submit(new Runnable() {

				@Override
				public void run() {
					handleMessage(messageListenerMap.get(topic));
				}

			});
		}
	}

	@SuppressWarnings({ "rawtypes", "unchecked" })
	public void handleMessage(MqMessageListener listener) {
		MqListener mqListener = listener.getClass().getAnnotation(MqListener.class);
		String topicName=SpringUtils.getConfigValue(mqListener.topic());
		String tagName=SpringUtils.getConfigValue(mqListener.tag());
		String queueUrl = amazonSqsAsync.getQueueUrl(topicName).getQueueUrl();
		while (true) {
			try {
				ReceiveMessageResult receiveMessageResult = amazonSqsAsync
						.receiveMessage(new ReceiveMessageRequest()
						.withQueueUrl(queueUrl)
						.withMaxNumberOfMessages(mqListener.maxNumberOfMessages())
						.withWaitTimeSeconds(mqListener.waitTimeSeconds()));
				
//				if (LOGGER.isInfoEnabled()) {
//					LOGGER.info("reviceing topic:{} size:{}", topic,receiveMessageResult.getMessages().size());
//				}
				
				if (receiveMessageResult.getMessages().isEmpty()) {
					try {
						Thread.sleep(mqListener.sleep());
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					continue;
				}
				
				for (Message message : receiveMessageResult.getMessages()) {
					
//					if (LOGGER.isInfoEnabled()) {
//						LOGGER.info("revice message topic:{} messageBody:{}", topic,message.getBody());
//					}
					
					SerializeConverterEditor editor = new SerializeConverterEditor(Object.class);
					editor.setValue(message.getBody());
					MqMessage am=(MqMessage)editor.converter();
					
					if("*".equals(tagName)||
							am.getTag().equals(tagName)) {
						listener.onMessage(am.getMessage());
						amazonSqsAsync.deleteMessage(queueUrl, message.getReceiptHandle());
					}
					
				}
			} catch (Exception e) {
				LoggerWrapper.print(e);
			}
		}
		
	}

}
